package com.atguigu.flink.streamapi.sink;

import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

/**
 * Created by Smexy on 2022/11/21
 *
 *      几乎用不到
 *
 *      案例：向Mysql写，统计，每种传感器，累积的最新的水位
 *
 CREATE TABLE `ws` (
 `id` varchar(100) NOT NULL,
 `ts` bigint(20) DEFAULT NULL,
 `vc` int(11) DEFAULT NULL,
 PRIMARY KEY (`id`)
 ) ENGINE=InnoDB DEFAULT CHARSET=utf8
 */
public class Demo3_CustomSink
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env
            .socketTextStream("hadoop103", 8888)
            .map(new MapFunction<String, WaterSensor>()
            {
                @Override
                public WaterSensor map(String value) throws Exception {
                    String[] data = value.split(",");
                    return new WaterSensor(
                        data[0],
                        Long.valueOf(data[1]),
                        Integer.valueOf(data[2])
                    );
                }
            })
            .keyBy(WaterSensor::getId)
            .reduce(new ReduceFunction<WaterSensor>()
            {

                @Override
                public WaterSensor reduce(WaterSensor value1, WaterSensor value2) throws Exception {
                    value2.setVc(value1.getVc() + value2.getVc());
                    return value2;
                }
            })
            .addSink(new MySink());

        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    /*
            自定义Sink
     */
    public static class MySink extends RichSinkFunction<WaterSensor>
    {

        private Connection connection;
        private PreparedStatement ps;

        @Override
        public void open(Configuration parameters) throws Exception {
            //创建连接
            connection = DriverManager.getConnection(
                "jdbc:mysql://hadoop104:3306/220704upm?useSSL=false&useUnicode=true&characterEncoding=utf8&rewriteBatchedStatements=true",
                "root",
                "000000"
            );

            ps = connection.prepareStatement(sql);
        }

        @Override
        public void close() throws Exception {
            if (connection != null){
                connection.close();
            }
        }

        //为了保证幂等性 replace into
        String sql = "replace into ws values(?,?,?)";
        //没来一个元素，执行一次
        @Override
        public void invoke(WaterSensor value, Context context) throws Exception {

                ps.setString(1,value.getId());
                ps.setLong(2,value.getTs());
                ps.setInt(3,value.getVc());

                //执行写入
            ps.execute();
        }
    }
}
